package com.rocketmq.demo.scheduled;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @ProjectName: rocketmq-demo
 * @Package: com.rocketmq.demo.scheduled
 * @ClassName: ScheduledMessageConsumer
 * @Author: mac
 * @Description:
 * @Date: 2021/7/1 10:12 下午
 * @Version: 1.0
 */
public class ScheduledMessageConsumer {

    public static void main(String[] args) {

        //实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");
        consumer.setNamesrvAddr("192.168.31.67:9876;192.168.31.126:9876;192.168.31.209:9876");

        try {
            //订阅topics
            consumer.subscribe("TestTopic","*");
            //注册消息监听者
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    for (MessageExt message : list) {
                        // Print approximate delay time period
                        System.out.println("Receive message[msgId=" + message.getMsgId() + "] " + (System.currentTimeMillis() - message.getStoreTimestamp()) + "ms later");
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            //启动消费者
            consumer.start();

        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
